package com.flute.haflute.net.messaging;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.InetAddress;
import java.util.Arrays;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;

import com.flute.haflute.common.SerializerInterface;

/**
 * 集群消息封装类
 * 
 */
public class Message {
	private static SerializerInterface<Message> serializer = new MessageSerializer();

	public int MAX_PACKET_SIZE = 1428;

	public static enum MessageType {
		HB, // 心跳
		HBACK, // 心跳确认
		HBACK2, // 心跳二次确认
		STATE_SYNC, // 集群状态同步
		STATE_CHG, // 节点状态改变
		JOB_NEW, JOB_ENDED, JOB_SCHEDULED, KEEP_ALIVE
	}

	public static enum Verb {
		HEARTBEARTING, STATECHANGE, TASK_NEW, TASK_PAUSE, TASK_END, JOB_NEW, JOB_SCHEDULE, JOB_ENDED, CLUSTER_MANAGE, KEEP_ALIVE
	}

	final Header header;
	private final byte[] body;

	public static SerializerInterface<Message> serializer() {
		return serializer;
	}

	public Message(Header header, byte[] body) {
		assert header != null;
		assert body != null;

		this.header = header;
		this.body = body;
	}

	public Message(InetAddress from, String messageType, Verb verb, byte[] body) {
		this(new Header(from, messageType, verb), body);
	}

	public byte[] getHeader(Object key) {
		return header.getDetail(key);
	}

	public void setHeader(String key, byte[] value) {
		header.setDetail(key, value);
	}

	public byte[] getMessageBody() {
		return body;
	}

	public InetAddress getFrom() {
		return header.getFrom();
	}

	public String getMessageType() {
		return header.getMessageType();
	}

	public String getMessageId() {
		return header.getMessageId();
	}

	void setMessageId(String id) {
		header.setMessageId(id);
	}

	Verb getVerb() {
		return header.getVerb();
	}

	void setVerb(Verb verb) {
		this.header.setVerb(verb);
	}

	public Message getReply(InetAddress from, byte[] args) {
		Header header = new Header(getMessageId(), from, MessageType.HB.name(),
				getVerb());
		return new Message(header, args);
	}

	public String toString() {
		StringBuilder sbuf = new StringBuilder("");
		String separator = System.getProperty("line.separator");
		sbuf.append("ID:" + getMessageId()).append(separator)
				.append("FROM:" + getFrom()).append(separator)
				.append("TYPE:" + getMessageType()).append(separator)
				.append("VERB:" + getVerb()).append(separator);
		return sbuf.toString();
	}

	/*
	 * @see java.lang.Object#hashCode()
	 */
	@Override
	public int hashCode() {
		final int prime = 31;
		int result = 1;
		result = prime * result + Arrays.hashCode(body);
		result = prime * result + ((header == null) ? 0 : header.hashCode());
		return result;
	}

	/*
	 * @see java.lang.Object#equals(java.lang.Object)
	 */
	@Override
	public boolean equals(Object obj) {
		if (this == obj)
			return true;
		if (obj == null)
			return false;
		if (getClass() != obj.getClass())
			return false;
		Message other = (Message) obj;
		if (!Arrays.equals(body, other.body))
			return false;
		if (header == null) {
			if (other.header != null)
				return false;
		} else if (!header.equals(other.header))
			return false;
		return true;
	}
}

class Header {
	private static SerializerInterface<Header> serializer = new HeaderSerializer();
	private static AtomicInteger idSeq = new AtomicInteger(0);

	static SerializerInterface<Header> serializer() {
		return serializer;
	}

	private InetAddress from;
	private String type;
	private String messageId;
	private Message.Verb verb;
	protected ConcurrentMap<String, byte[]> details = new ConcurrentHashMap<String, byte[]>();

	Header(String id, InetAddress from, String messageType, Message.Verb verb) {
		assert id != null;
		assert from != null;
		assert messageType != null;
		assert verb != null;

		this.messageId = id;
		this.from = from;
		this.type = messageType;
		this.verb = verb;
	}

	Header(String id, InetAddress from, String messageType, Message.Verb verb,
			ConcurrentMap<String, byte[]> headDetails) {
		this(id, from, messageType, verb);
		this.details = headDetails;
	}

	Header(InetAddress from, String messageType, Message.Verb verb) {
		this(Integer.toString(idSeq.incrementAndGet()), from, messageType, verb);
	}

	InetAddress getFrom() {
		return from;
	}

	String getMessageType() {
		return type;
	}

	String getMessageId() {
		return messageId;
	}

	void setMessageId(String id) {
		messageId = id;
	}

	byte[] getDetail(Object key) {
		return details.get(key);
	}

	void setDetail(String key, byte[] value) {
		details.put(key, value);
	}

	public Message.Verb getVerb() {
		return verb;
	}

	public void setVerb(Message.Verb verb) {
		this.verb = verb;
	}

	/*
	 * @see java.lang.Object#hashCode()
	 */
	@Override
	public int hashCode() {
		final int prime = 31;
		int result = 1;
		result = prime * result + ((details == null) ? 0 : details.hashCode());
		result = prime * result + ((from == null) ? 0 : from.hashCode());
		result = prime * result
				+ ((messageId == null) ? 0 : messageId.hashCode());
		result = prime * result + ((type == null) ? 0 : type.hashCode());
		result = prime * result + ((verb == null) ? 0 : verb.hashCode());
		return result;
	}

	/*
	 * @see java.lang.Object#equals(java.lang.Object)
	 */
	@Override
	public boolean equals(Object obj) {
		if (this == obj)
			return true;
		if (obj == null)
			return false;
		if (getClass() != obj.getClass())
			return false;
		Header other = (Header) obj;
		if (details == null) {
			if (other.details != null)
				return false;
		} else if (!details.equals(other.details))
			return false;
		if (from == null) {
			if (other.from != null)
				return false;
		} else if (!from.equals(other.from))
			return false;
		if (messageId == null) {
			if (other.messageId != null)
				return false;
		} else if (!messageId.equals(other.messageId))
			return false;
		if (type == null) {
			if (other.type != null)
				return false;
		} else if (!type.equals(other.type))
			return false;
		if (verb == null) {
			if (other.verb != null)
				return false;
		} else if (!verb.equals(other.verb))
			return false;
		return true;
	}
}

class HeaderSerializer implements SerializerInterface<Header> {
	public void serialize(Header t, DataOutputStream dos) throws IOException {
		dos.writeUTF(t.getMessageId());
		serializeAddress(t.getFrom(), dos);
		dos.writeUTF(t.getMessageType());
		dos.writeInt(t.getVerb().ordinal());

		/* Serialize the message header */
		int size = t.details.size();
		dos.writeInt(size);
		Set<String> keys = t.details.keySet();

		for (String key : keys) {
			dos.writeUTF(key);
			byte[] value = t.details.get(key);
			dos.writeInt(value.length);
			dos.write(value);
		}
	}

	public Header deserialize(DataInputStream dis) throws IOException {
		String id = dis.readUTF();
		InetAddress from = deserializeAddress(dis);
		String type = dis.readUTF();
		int verbOrdinal = dis.readInt();

		// de-serialize the message header
		int size = dis.readInt();
		ConcurrentMap<String, byte[]> details = new ConcurrentHashMap<String, byte[]>(
				size);
		for (int i = 0; i < size; ++i) {
			String key = dis.readUTF();
			int length = dis.readInt();
			byte[] bytes = new byte[length];
			dis.readFully(bytes);
			details.put(key, bytes);
		}

		return new Header(id, from, type, Message.Verb.values()[verbOrdinal],
				details);
	}

	private void serializeAddress(InetAddress nodeIp, DataOutputStream dos)
			throws IOException {
		byte[] buf = nodeIp.getAddress();
		dos.writeByte(buf.length);
		dos.write(buf);
	}

	private InetAddress deserializeAddress(DataInputStream dis)
			throws IOException {
		byte[] bytes = new byte[dis.readByte()];
		dis.readFully(bytes, 0, bytes.length);
		return InetAddress.getByAddress(bytes);
	}
}

class MessageSerializer implements SerializerInterface<Message> {
	public void serialize(Message t, DataOutputStream dos) throws IOException {
		Header.serializer().serialize(t.header, dos);
		byte[] bytes = t.getMessageBody();
		dos.writeInt(bytes.length);
		dos.write(bytes);
	}

	public Message deserialize(DataInputStream dis) throws IOException {
		Header header = Header.serializer().deserialize(dis);
		int size = dis.readInt();
		byte[] bytes = new byte[size];
		dis.readFully(bytes);
		return new Message(header, bytes);
	}
}
